package com.ev.custom.rabbit.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.ev.common.jobs.BackupsDBJob;
import com.ev.custom.domain.DeviceDataDO;
import com.ev.custom.domain.NoticeDO;
import com.ev.custom.service.NoticeService;
import com.ev.custom.service.WeChatService;
import com.ev.custom.vo.NoticeEntity;
import com.ev.framework.socket.CusWebSocket;
import com.ev.framework.utils.InfluxDbUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import org.influxdb.InfluxDB;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.text.ParseException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;

//@Component
public class MqListener {
    private static final Logger log= LoggerFactory.getLogger(MqListener.class);

//    @Autowired
//    private ObjectMapper objectMapper;
//
//    @Autowired
//    private NoticeService noticeService;
//
//    @Autowired
//    private Environment env;
//
//    @Autowired
//    private WeChatService weChatService;
//
//    @RabbitListener(queues = "${data.device.queue.name}",containerFactory = "singleListenerContainer")
//    public void consumeUserLogQueue(@Payload byte[] message){
//        try {
//            List<LinkedHashMap> deviceDatas=objectMapper.readValue(message, ArrayList.class);
//            saveInfluxData(deviceDatas,"device_electricity");
//        }catch (Exception e){
//            e.printStackTrace();
//        }
//    }
//
//    /**
//     * 保存数据到时序数据库
//     */
//    private void saveInfluxData(List<LinkedHashMap> deviceDatas, String measurement) {
//        InfluxDbUtils influxDB = InfluxDbUtils.setUp();
//        List<String> records = new ArrayList<String>();
//        for(LinkedHashMap dataMap : deviceDatas){
//            if(!StringUtils.isEmpty(dataMap.get("value").toString().trim())){
//                try{
//                    Map<String, String> tags = new HashMap<String, String>();
//                    tags.put("point", dataMap.get("pointId").toString());
//                    Map<String, Object> values = new HashMap<String, Object>();
//                    values.put("value", Double.parseDouble(dataMap.get("value").toString().trim()));
//                    values.put("valueRecord", Double.parseDouble(dataMap.get("valueRecord").toString().trim()));
//                    Point point = influxDB.pointBuilder(measurement, tags, values,dataMap.get("time").toString());
//                    BatchPoints batchPoint = BatchPoints.database(measurement).consistency(InfluxDB.ConsistencyLevel.ALL).build();
//                    batchPoint.point(point);
//                    records.add(batchPoint.lineProtocol());
//                }catch (Exception e){
//                    continue;
//                }
//            }
//
//        }
//        // 将两条数据批量插入到数据库中
//        influxDB.batchInsert("sentinel_log", null, InfluxDB.ConsistencyLevel.ALL, records);
//    }
//
//    @RabbitListener(queues = "gyhl.data.notice.queue",containerFactory = "singleListenerContainer")
//    public void consumeNoticeQueue(@Payload byte[] message){
//        try {
//            NoticeEntity notice=objectMapper.readValue(message, NoticeEntity.class);
//            saveAndSendNotice(notice);
//        }catch (Exception e){
//            e.printStackTrace();
//        }
//    }
//
//    /**
//     * 发送微信卡片消息并发送websocket消息
//     */
//    private void saveAndSendNotice(NoticeEntity notice) throws IOException, ParseException {
//        weChatService.sendTextCardMessage(notice.getNoticeDO(),notice.getToUserList());
//        for(Long toUserId: notice.getToUserList()){
//            CusWebSocket.sendInfo(notice.toString(),toUserId.toString());
//        }
//    }
//
//    @RabbitListener(queues = "gyhl.data.db.queue", containerFactory = "singleListenerContainer")
//    public void consumeDbQueue(@Payload byte[] message) {
//        try {
//            boolean success = BackupsDBJob.backUpDB(env);
//            // 若不成功则推送微信消息
//            if (!success) {
//                NoticeEntity notice = objectMapper.readValue(message, NoticeEntity.class);
//                NoticeDO noticeDO = notice.getNoticeDO();
//                noticeService.saveAndSendSocket(noticeDO.getTitle(),noticeDO.getContent(),null,noticeDO.getContentDetail(),7L,noticeDO.getFromUserId(),notice.getToUserList());
//            }
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//
//
//    @RabbitListener(queues = "data.device.treated",containerFactory = "batchQueueRabbitListenerContainerFactory")
//    public void directQueue(@Header(AmqpHeaders.CHANNEL) Channel channel,@Payload byte[] msg, Message message) throws Exception {
//        try {
//            String info = new String(msg);
////            System.out.println(info);
//            JSONArray datas = (JSONArray) JSON.parseObject(info).get("datas");
//            List<DeviceDataDO> listInfluxData = JSONObject.parseArray(datas.toJSONString(),DeviceDataDO.class);
//            //将数据按照时间降序排序并获取每个测点的最新数据
//            List<DeviceDataDO> filterList = listInfluxData.stream()
//                    .sorted(Comparator.comparing(DeviceDataDO::getTime).reversed()).filter(distinctByKey(e -> e.getPointId()))
//                    .collect(Collectors.toList());
//            // 发送websoket 消息
//            CusWebSocket.sendInfo("{\"datas\":"+ JSON.toJSONString(filterList) +"}");
//            // 保存入influxdb数据库
//            newSaveInfluxData(filterList,"device_electricity");
//            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//        }catch (Exception e){
//            e.printStackTrace();
//        }
//    }
//
//    /**
//     * 保存数据到时序数据库
//     */
//    private void newSaveInfluxData(List<DeviceDataDO> deviceDatas, String measurement) {
//        InfluxDbUtils influxDB = InfluxDbUtils.setUp();
//        List<String> records = new ArrayList<String>();
//        for(DeviceDataDO dataMap : deviceDatas){
//            if(!StringUtils.isEmpty(dataMap.getValue().trim())){
//                try{
//                    Map<String, String> tags = new HashMap<String, String>();
//                    tags.put("point", dataMap.getPointId());
//                    Map<String, Object> values = new HashMap<String, Object>();
//                    values.put("value", Double.parseDouble(dataMap.getValue().toString().trim()));
//                    values.put("valueRecord", Double.parseDouble(dataMap.getValueRecord().toString().trim()));
//                    Point point = influxDB.pointBuilder(measurement, tags, values,dataMap.getTime().toString());
//                    BatchPoints batchPoint = BatchPoints.database(measurement).consistency(InfluxDB.ConsistencyLevel.ALL).build();
//                    batchPoint.point(point);
//                    records.add(batchPoint.lineProtocol());
//                }catch (Exception e){
//                    continue;
//                }
//            }
//
//        }
//        // 将两条数据批量插入到数据库中
//        influxDB.batchInsert("sentinel_log", null, InfluxDB.ConsistencyLevel.ALL, records);
//    }
//
//    private static <T> Predicate<T> distinctByKey(Function<? super T, ?> keyExtractor) {
//        Map<Object,Boolean> seen = new ConcurrentHashMap<>();
//        return t -> seen.putIfAbsent(keyExtractor.apply(t), Boolean.TRUE) == null;
//    }
}
